package com.lagou.kafka.demo;

import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.Collections;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;

/**
 * @param
 * @author oyz
 * @version 1.0.4
 * @description: TODO kafka客户端
 * @date 2022/6/14 21:12
 */
public class MyAdminClient  {

    private KafkaAdminClient kafkaAdminClient;


    @Before
    public void before(){
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.106.129:9092");
        properties.setProperty(ConsumerConfig.CLIENT_ID_CONFIG,"admin_001");

        kafkaAdminClient = (KafkaAdminClient)KafkaAdminClient.create(properties);
    }


    @After
    public void after(){
        kafkaAdminClient.close();
    }


    @Test
    public void testListTopics() throws ExecutionException, InterruptedException {
        Set<Integer> singleton = Collections.singleton(0);
        /*singleton.add(1);
        singleton.add(2);*/
        singleton.forEach(o->{
            System.out.println(o);
        });
        /*// 列出主题
        ListTopicsResult listTopicsResult = kafkaAdminClient.listTopics();
        // 将请求变为同步的,直接获取请求结果
        Set<String> strings = listTopicsResult.names().get();
        strings.forEach(name->{
            System.out.println(name);
        });*/
    }

}
